package com.king.first.app.windows;

import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RichAggregateFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

/**
 * 增量聚合函数
 */
public class Window_Function_All {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> socketTextStream = env.socketTextStream("hadoop01", 9999);
        SingleOutputStreamOperator<Tuple2<String, Integer>> flatMapStream = socketTextStream.flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (v, collector) -> {
            String[] s = v.split(" ");
            for (String s1 : s) {
                collector.collect(Tuple2.of(s1, 1));
            }
        }).returns(Types.TUPLE(Types.STRING, Types.INT));

        AllWindowedStream<Tuple2<String, Integer>, TimeWindow> allWindowedStream = flatMapStream.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(3)));
        SingleOutputStreamOperator<Tuple2<String, Integer>> aggregate1 = allWindowedStream.aggregate(new AggregateFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple2<String, Integer>>() {

            @Override
            public Tuple2<String, Integer> createAccumulator() {
                return new Tuple2<String, Integer>("", 0);
            }

            @Override
            public Tuple2<String, Integer> add(Tuple2<String, Integer> tuple2, Tuple2<String, Integer> tuple22) {
                Integer count = 0;
                if ("".equals(tuple22.f0)){
                    count += tuple2.f1;
                } else {
                    count = tuple2.f1 + tuple22.f1;
                }
                return Tuple2.of(tuple2.f0, count);
            }

            @Override
            public Tuple2<String, Integer> getResult(Tuple2<String, Integer> tuple2) {
                return tuple2;
            }

            @Override
            public Tuple2<String, Integer> merge(Tuple2<String, Integer> tuple2, Tuple2<String, Integer> acc1) {
                return Tuple2.of(tuple2.f0, tuple2.f1 + acc1.f1);
            }
        });
        SingleOutputStreamOperator<Tuple2<String, Integer>> aggregate = allWindowedStream.aggregate(new AggregateFunction<Tuple2<String, Integer>, Integer, Tuple2<String, Integer>>() {
            private String key = "";

            @Override
            public Integer createAccumulator() {
                return 0;
            }

            @Override
            public Integer add(Tuple2<String, Integer> tuple2, Integer integer) {
                key = tuple2.f0;
                return tuple2.f1 + integer;
            }

            @Override
            public Tuple2<String, Integer> getResult(Integer integer) {
                return Tuple2.of(key, integer);
            }

            @Override
            public Integer merge(Integer integer, Integer acc1) {
                return integer + acc1;
            }
        });

        KeyedStream<Tuple2<String, Integer>, String> keyedStream = aggregate1.keyBy(v -> v.f0);
        keyedStream.sum(1).print();
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
